package com.changfen.produce;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.changfen.event.Event;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author changfeng.chen@things-matrix.com
 * @date 2022/8/27 11:33
 **/
@Component
public class DisruptorExecutor implements InitializingBean, Produce {
    @Resource
    private List<EventHandler> eventHandlerList;

    private Disruptor<Event> disruptor;

    @PostConstruct
    public void createDisruptor() {
        disruptor = new Disruptor(Event::new, 1024, new ThreadFactory() {
            private final AtomicLong count = new AtomicLong(0L);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("[disruptor worker num = {" + this.count.getAndIncrement() + "}]");
                return thread;
            }
        });
    }

    @PreDestroy
    public void destroy() {
        disruptor.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        disruptor.handleEventsWith(eventHandlerList.toArray(new EventHandler[]{}));
        disruptor.start();
    }

    @Override
    public void publishEvent(Event event1) {
        // Get the ring buffer from the Disruptor to be used for publishing.
        RingBuffer<Event> ringBuffer = disruptor.getRingBuffer();
        ringBuffer.publishEvent((event, sequence) -> {
            try {
                BeanUtils.copyProperties(event1, event);
            } catch (Exception e) {
                e.printStackTrace();
            }
            event.setValue(sequence);
        });

    }
}
